/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.zookeeper;

import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.lang.Thread.UncaughtExceptionHandler;
import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.util.LinkedList;
import java.util.List;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.LinkedBlockingQueue;

import javax.security.auth.login.LoginException;
import javax.security.sasl.SaslException;

import org.apache.jute.BinaryInputArchive;
import org.apache.jute.BinaryOutputArchive;
import org.apache.jute.Record;
import org.apache.zookeeper.AsyncCallback.ACLCallback;
import org.apache.zookeeper.AsyncCallback.Children2Callback;
import org.apache.zookeeper.AsyncCallback.ChildrenCallback;
import org.apache.zookeeper.AsyncCallback.DataCallback;
import org.apache.zookeeper.AsyncCallback.StatCallback;
import org.apache.zookeeper.AsyncCallback.StringCallback;
import org.apache.zookeeper.AsyncCallback.VoidCallback;
import org.apache.zookeeper.Watcher.Event;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZooDefs.OpCode;
import org.apache.zookeeper.ZooKeeper.States;
import org.apache.zookeeper.ZooKeeper.WatchRegistration;
import org.apache.zookeeper.client.HostProvider;
import org.apache.zookeeper.client.ZooKeeperSaslClient;
import org.apache.zookeeper.proto.AuthPacket;
import org.apache.zookeeper.proto.ConnectRequest;
import org.apache.zookeeper.proto.CreateResponse;
import org.apache.zookeeper.proto.ExistsResponse;
import org.apache.zookeeper.proto.GetACLResponse;
import org.apache.zookeeper.proto.GetChildren2Response;
import org.apache.zookeeper.proto.GetChildrenResponse;
import org.apache.zookeeper.proto.GetDataResponse;
import org.apache.zookeeper.proto.GetSASLRequest;
import org.apache.zookeeper.proto.ReplyHeader;
import org.apache.zookeeper.proto.RequestHeader;
import org.apache.zookeeper.proto.SetACLResponse;
import org.apache.zookeeper.proto.SetDataResponse;
import org.apache.zookeeper.proto.SetWatches;
import org.apache.zookeeper.proto.WatcherEvent;
import org.apache.zookeeper.server.ByteBufferInputStream;
import org.apache.zookeeper.server.ZooTrace;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * This class manages the socket i/o for the client. ClientCnxn maintains a list
 * of available servers to connect to and "transparently" switches servers it is
 * connected to as needed.
 *
 */
public class ClientCnxn {
    private static final Logger LOG = LoggerFactory.getLogger(ClientCnxn.class);

    /** This controls whether automatic watch resetting is enabled.
     * Clients automatically reset watches during session reconnect, this
     * option allows the client to turn off this behavior by setting
     * the environment variable "zookeeper.disableAutoWatchReset" to "true" */
    private static boolean disableAutoWatchReset;

    static {
        disableAutoWatchReset = Boolean.getBoolean("zookeeper.disableAutoWatchReset");
    }

    static class AuthData {
        AuthData(String scheme, byte data[]) {
            this.scheme = scheme;
            this.data = data;
        }

        String scheme;

        byte data[];
    }

    private final CopyOnWriteArraySet<AuthData> authInfo = new CopyOnWriteArraySet<AuthData>();

    // Packet:封装了客户端一次请求或服务端一次响应的完整数据
    // 已经发送但是等待服务端响应的packet集合
    private final LinkedList<Packet> pendingQueue = new LinkedList<Packet>();

    /** 需要发送的packet集合 */
    private final LinkedList<Packet> outgoingQueue = new LinkedList<Packet>();
    // 建立连接的超时时间
    private int connectTimeout;

    /** 服务端认为的下一次会话过期的具体时间 */
    private volatile int negotiatedSessionTimeout;

    /** 客户端认为的最大会话超时时间，默认为sessionTimeout * 2 / 3 */
    private int readTimeout;
    // 会话的超时时间
    private final int sessionTimeout;

    private final ZooKeeper zooKeeper;

    private final ClientWatchManager watcher;

    private long sessionId;

    private byte sessionPasswd[] = new byte[16];

    /**
     * If true, the connection is allowed to go to r-o mode. This field's value
     * is sent, besides other data, during session creation handshake. If the
     * server on the other side of the wire is partitioned it'll accept
     * read-only clients only.
     */
    private boolean readOnly;

    // 客户端的命名空间，客户端所有的数据节点的路径都会默认在这层路径下创建。可通过`connectString`参数
    // 传入，如`127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002/app/a`，则`chrootPath=/app/a
    final String chrootPath;

    /** 负责与服务端之间的读写网络通信和心跳维持。 */
    final SendThread sendThread;
    /** 对经过SendThread反序列化后的具体请求根据类型的不同，具体处理，同时也会负责对请求事件的watch回调 */
    final EventThread eventThread;

    /**
     * Set to true when close is called. Latches the connection such that we
     * don't attempt to re-connect to the server if in the middle of closing the
     * connection (client sends session disconnect to server as part of close
     * operation)
     */
    private volatile boolean closing = false;
    
    /** 在会话未过期前与当前服务端连接中断时，自动选择一个可用的服务端重建连接 */
    private final HostProvider hostProvider;

    /**
     * Is set to true when a connection to a r/w server is established for the
     * first time; never changed afterwards.
     * <p>
     * Is used to handle situations when client without sessionId connects to a
     * read-only server. Such client receives "fake" sessionId from read-only
     * server, but this sessionId is invalid for other servers. So when such
     * client finds a r/w server, it sends 0 instead of fake sessionId during
     * connection handshake and establishes new, valid session.
     * <p>
     * If this field is false (which implies we haven't seen r/w server before)
     * then non-zero sessionId is fake, otherwise it is valid.
     */
    volatile boolean seenRwServerBefore = false;


    public ZooKeeperSaslClient zooKeeperSaslClient;

    public long getSessionId() {
        return sessionId;
    }

    public byte[] getSessionPasswd() {
        return sessionPasswd;
    }

    public int getSessionTimeout() {
        return negotiatedSessionTimeout;
    }

    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder();

        SocketAddress local = sendThread.getClientCnxnSocket().getLocalSocketAddress();
        SocketAddress remote = sendThread.getClientCnxnSocket().getRemoteSocketAddress();
        sb
            .append("sessionid:0x").append(Long.toHexString(getSessionId()))
            .append(" local:").append(local)
            .append(" remoteserver:").append(remote)
            .append(" lastZxid:").append(lastZxid)
            .append(" xid:").append(xid)
            .append(" sent:").append(sendThread.getClientCnxnSocket().getSentCount())
            .append(" recv:").append(sendThread.getClientCnxnSocket().getRecvCount())
            .append(" queuedpkts:").append(outgoingQueue.size())
            .append(" pendingresp:").append(pendingQueue.size())
            .append(" queuedevents:").append(eventThread.waitingEvents.size());

        return sb.toString();
    }

    /**
     * Packet是ClientCnxn内部定义的一个堆协议层的封装，用作Zookeeper中请求和响应的载体。
     * Packet包含了请求头（requestHeader）、响应头（replyHeader）、请求体（request）、
     *      响应体（response）、节点路径（clientPath/serverPath）、
     *      注册的Watcher（watchRegistration）等信息
     *
     * 然而，并非Packet中所有的属性都在客户端与服务端之间进行网络传输，只会将requestHeader、request、readOnly三个属性序列化，
     * 并生成可用于底层网络传输的ByteBuffer， 其他属性都保存在客户端的上下文中，不会进行与服务端之间的网络传输。
     *
     */
    static class Packet {
        /** 请求头 */
        RequestHeader requestHeader;
        /** 响应头 */
        ReplyHeader replyHeader;
        /** 请求体 */
        Record request;
        /** 响应体 */
        Record response;
        /** 序列化之后的byteBuffer */
        ByteBuffer bb;

        /** client节点路径，不含chrootPath */
        String clientPath;
        /** server节点路径,含chrootPath */
        String serverPath;
        /** 是否结束(已经得到响应才能结束) */
        boolean finished;
        /** 异步回调 */
        AsyncCallback cb;
        /** 上下文 */
        Object ctx;
        /** 注册的watcher */
        WatchRegistration watchRegistration;
        /** 只读 */
        public boolean readOnly;

        /** Convenience ctor */
        Packet(RequestHeader requestHeader, ReplyHeader replyHeader, Record request, Record response, WatchRegistration watchRegistration) {
            this(requestHeader, replyHeader, request, response, watchRegistration, false);
        }

        Packet(RequestHeader requestHeader, ReplyHeader replyHeader, Record request, Record response, WatchRegistration watchRegistration, boolean readOnly) {

            this.requestHeader = requestHeader;
            this.replyHeader = replyHeader;
            this.request = request;
            this.response = response;
            this.readOnly = readOnly;
            this.watchRegistration = watchRegistration;
        }

        /**
         *  序列化创建byteBuffer记录在bb字段中
         */
        public void createBB() {
            try {
                ByteArrayOutputStream baos = new ByteArrayOutputStream();
                BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
                boa.writeInt(-1, "len");
                if (requestHeader != null) {
                    // 序列化请求头，包含xid和type
                    requestHeader.serialize(boa, "header");
                }
                if (request instanceof ConnectRequest) {
                    request.serialize(boa, "connect");
                    boa.writeBool(readOnly, "readOnly");
                } else if (request != null) {
                    // 序列化request(对于特定请求如GetDataRequest,包含了是否存在watcher的标志位)
                    request.serialize(boa, "request");
                }
                baos.close();
                this.bb = ByteBuffer.wrap(baos.toByteArray());
                this.bb.putInt(this.bb.capacity() - 4);
                this.bb.rewind();
            } catch (IOException e) {
                LOG.warn("Ignoring unexpected exception", e);
            }
        }

        @Override
        public String toString() {
            StringBuilder sb = new StringBuilder();

            sb.append("clientPath:" + clientPath);
            sb.append(" serverPath:" + serverPath);
            sb.append(" finished:" + finished);

            sb.append(" header:: " + requestHeader);
            sb.append(" replyHeader:: " + replyHeader);
            sb.append(" request:: " + request);
            sb.append(" response:: " + response);

            // jute toString is horrible, remove unnecessary newlines
            return sb.toString().replaceAll("\r*\n+", " ");
        }
    }

    /**
     * Creates a connection object. The actual network connect doesn't get
     * established until needed. The start() instance method must be called
     * subsequent to construction.
     *
     * @param chrootPath - the chroot of this client. Should be removed from this Class in ZOOKEEPER-838
     * @param hostProvider the list of ZooKeeper servers to connect to
     * @param sessionTimeout  the timeout for connections.
     * @param zooKeeper  the zookeeper object that this connection is related to.
     * @param watcher watcher for this connection
     * @param clientCnxnSocket the socket implementation used (e.g. NIO/Netty)
     * @param canBeReadOnly whether the connection is allowed to go to read-only mode in case of partitioning
     * @throws IOException
     */
    public ClientCnxn(String chrootPath, HostProvider hostProvider, int sessionTimeout, ZooKeeper zooKeeper,
            ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket, boolean canBeReadOnly)   throws IOException {
        this(chrootPath, hostProvider, sessionTimeout, zooKeeper, watcher, clientCnxnSocket, 0, new byte[16], canBeReadOnly);
    }

    /**
     * Creates a connection object. The actual network connect doesn't get
     * established until needed. The start() instance method must be called
     * subsequent to construction.
     *
     * @param chrootPath - the chroot of this client. Should be removed from this Class in ZOOKEEPER-838
     * @param hostProvider the list of ZooKeeper servers to connect to
     * @param sessionTimeout the timeout for connections.
     * @param zooKeeper the zookeeper object that this connection is related to.
     * @param watcher watcher for this connection
     * @param clientCnxnSocket the socket implementation used (e.g. NIO/Netty)
     * @param sessionId session id if re-establishing session
     * @param sessionPasswd session passwd if re-establishing session
     * @param canBeReadOnly whether the connection is allowed to go to read-only mode in case of partitioning
     * @throws IOException
     */
    public ClientCnxn(String chrootPath, HostProvider hostProvider, int sessionTimeout, ZooKeeper zooKeeper,
            ClientWatchManager watcher, ClientCnxnSocket clientCnxnSocket, long sessionId, byte[] sessionPasswd, boolean canBeReadOnly) {
        this.zooKeeper = zooKeeper;
        this.watcher = watcher;
        this.sessionId = sessionId;
        this.sessionPasswd = sessionPasswd;
        this.sessionTimeout = sessionTimeout;
        this.hostProvider = hostProvider;
        this.chrootPath = chrootPath;

        connectTimeout = sessionTimeout / hostProvider.size(); // 连接超时时间
        readTimeout = sessionTimeout * 2 / 3; // 读取数据超时时间
        readOnly = canBeReadOnly; //是否为只读客户端，默认不是

        sendThread = new SendThread(clientCnxnSocket); // 基于底层的socket
        eventThread = new EventThread(); // 专门接收zk发送过来的回调事件

    }

    /**
     * tests use this to check on reset of watches
     * @return if the auto reset of watches are disabled
     */
    public static boolean getDisableAutoResetWatch() {
        return disableAutoWatchReset;
    }
    /**
     * tests use this to set the auto reset
     * @param b the value to set disable watches to
     */
    public static void setDisableAutoResetWatch(boolean b) {
        disableAutoWatchReset = b;
    }
    public void start() {
        sendThread.start();
        eventThread.start();
    }

    private Object eventOfDeath = new Object();

    private final static UncaughtExceptionHandler uncaughtExceptionHandler = new UncaughtExceptionHandler() {
        @Override
        public void uncaughtException(Thread t, Throwable e) {
            LOG.error("from " + t.getName(), e);
        }
    };

    private static class WatcherSetEventPair {
        private final Set<Watcher> watchers;
        private final WatchedEvent event;

        public WatcherSetEventPair(Set<Watcher> watchers, WatchedEvent event) {
            this.watchers = watchers;
            this.event = event;
        }
    }

    /**
     * Guard against creating "-EventThread-EventThread-EventThread-..." thread
     * names when ZooKeeper object is being created from within a watcher.
     * See ZOOKEEPER-795 for details.
     */
    private static String makeThreadName(String suffix) {
        String name = Thread.currentThread().getName().replaceAll("-EventThread", "");
        return name + suffix;
    }

    class EventThread extends Thread {
       private final LinkedBlockingQueue<Object> waitingEvents = new LinkedBlockingQueue<Object>();
       private volatile KeeperState sessionState = KeeperState.Disconnected;

       private volatile boolean wasKilled = false;
       private volatile boolean isRunning = false;

        EventThread() {
            super(makeThreadName("-EventThread"));
            setUncaughtExceptionHandler(uncaughtExceptionHandler);
            setDaemon(true);
        }

        public void queueEvent(WatchedEvent event) {
            if (event.getType() == EventType.None && sessionState == event.getState()) {
                return;
            }
            sessionState = event.getState();
            // 调用 materialize方法
            WatcherSetEventPair pair = new WatcherSetEventPair( watcher.materialize(event.getState(), event.getType(), event.getPath()), event);
            waitingEvents.add(pair);
        }

       public void queuePacket(Packet packet) {
          if (wasKilled) {
             synchronized (waitingEvents) {
                if (isRunning) {
                    waitingEvents.add(packet);
                } else {
                    processEvent(packet);
                }
             }
          } else {
             waitingEvents.add(packet);
          }
       }

        public void queueEventOfDeath() {
            waitingEvents.add(eventOfDeath);
        }

        // 从waitinEvents这个队列中获取事件，然后进行相应的处理
        @Override
        public void run() {
           try {
              isRunning = true;
              while (true) {
                 Object event = waitingEvents.take(); //获取对应事件
                 if (event == eventOfDeath) {
                    wasKilled = true;
                 } else {
                     // 核心方法, 处理事件，对notification调用对应的watcher操作
                    processEvent(event);
                 }
                 if (wasKilled) {
                     synchronized (waitingEvents) {
                        if (waitingEvents.isEmpty()) {
                           isRunning = false;
                           break;
                        }
                     }
                 }
              }
           } catch (InterruptedException e) {
              LOG.error("Event thread exiting due to interruption", e);
           }

            LOG.info("EventThread shut down");
        }

       private void processEvent(Object event) {
          try {
              // watcher监听事件
              if (event instanceof WatcherSetEventPair) {
                  WatcherSetEventPair pair = (WatcherSetEventPair) event;
                  // 循环watcher集合，调用watcher的回调方法(WatchedEvent event)处理该事件
                  /**
                   * getData("/brokers", new Watcher(){
                   *     @Override
                   *     public void process(WatchedEvent event) {
                   *         // 监听器回调 处理逻辑
                   *     }
                   * }, null);
                   */
                  for (Watcher watcher : pair.watchers) {
                      try {
                          // 还记得Watcher接口中对于watcher实现类都要实现的一个process(WatchedEvent event)方法，就是这个
                          watcher.process(pair.event);
                      } catch (Throwable t) {
                          LOG.error("Error while calling watcher ", t);
                      }
                  }
              }
              // 异步回调
              else {
                  Packet p = (Packet) event;
                  // ResultCode服务器端响应码，常见码如下
                  // 0：接口调用成功 -4客户端与服务器端连接已断开
                  // -110指定节点已存在 -112会话已过期
                  int rc = 0;
                  String clientPath = p.clientPath;
                  if (p.replyHeader.getErr() != 0) {
                      rc = p.replyHeader.getErr();
                  }
                  if (p.cb == null) {
                      LOG.warn("Somehow a null cb got to EventThread!");
                  }
                  // 如果是exists/setData/setAcl命令异步方式的服务器端响应
                  else if (p.response instanceof ExistsResponse
                                          || p.response instanceof SetDataResponse
                                          || p.response instanceof SetACLResponse) {
                      StatCallback cb = (StatCallback) p.cb;
                      //rc==0表示接口调用成功
                      if (rc == 0) {
                          if (p.response instanceof ExistsResponse) {
                              cb.processResult(rc, clientPath, p.ctx, ((ExistsResponse) p.response).getStat());
                          } else if (p.response instanceof SetDataResponse) {
                              cb.processResult(rc, clientPath, p.ctx, ((SetDataResponse) p.response).getStat());
                          } else if (p.response instanceof SetACLResponse) {
                              cb.processResult(rc, clientPath, p.ctx, ((SetACLResponse) p.response).getStat());
                          }
                      } else {
                          cb.processResult(rc, clientPath, p.ctx, null);
                      }
                  } else if (p.response instanceof GetDataResponse) {
                      DataCallback cb = (DataCallback) p.cb;
                      GetDataResponse rsp = (GetDataResponse) p.response;
                      //rc==0表示接口调用成功
                      if (rc == 0) {
                          cb.processResult(rc, clientPath, p.ctx, rsp.getData(), rsp.getStat());
                      } else {
                          cb.processResult(rc, clientPath, p.ctx, null, null);
                      }
                  } else if (p.response instanceof GetACLResponse) {
                      ACLCallback cb = (ACLCallback) p.cb;
                      GetACLResponse rsp = (GetACLResponse) p.response;
                      if (rc == 0) {
                          cb.processResult(rc, clientPath, p.ctx, rsp.getAcl(), rsp.getStat());
                      } else {
                          cb.processResult(rc, clientPath, p.ctx, null,  null);
                      }
                  } else if (p.response instanceof GetChildrenResponse) {
                      ChildrenCallback cb = (ChildrenCallback) p.cb;
                      GetChildrenResponse rsp = (GetChildrenResponse) p.response;
                      if (rc == 0) {
                          cb.processResult(rc, clientPath, p.ctx, rsp.getChildren());
                      } else {
                          cb.processResult(rc, clientPath, p.ctx, null);
                      }
                  } else if (p.response instanceof GetChildren2Response) {
                      Children2Callback cb = (Children2Callback) p.cb;
                      GetChildren2Response rsp = (GetChildren2Response) p.response;
                      if (rc == 0) {
                          cb.processResult(rc, clientPath, p.ctx, rsp.getChildren(), rsp.getStat());
                      } else {
                          cb.processResult(rc, clientPath, p.ctx, null, null);
                      }
                  } else if (p.response instanceof CreateResponse) {
                      StringCallback cb = (StringCallback) p.cb;
                      CreateResponse rsp = (CreateResponse) p.response;
                      if (rc == 0) {
                          cb.processResult(rc, clientPath, p.ctx,  (chrootPath == null  ? rsp.getPath()  : rsp.getPath().substring(chrootPath.length())));
                      } else {
                          cb.processResult(rc, clientPath, p.ctx, null);
                      }
                  } else if (p.cb instanceof VoidCallback) {
                      VoidCallback cb = (VoidCallback) p.cb;
                      cb.processResult(rc, clientPath, p.ctx);
                  }
              }
          } catch (Throwable t) {
              LOG.error("Caught unexpected throwable", t);
          }
       }
    }

    /**
     * Packet请求发送，收到回复，进行处理之后
     *
     * 很清晰，getData,getChildren, exists 请求先发出去
     * 直到这些请求成功完成了，返回了响应，此时调用finishPacket方法
     * 在这里，才会进行监听器的注册
     */
    private void finishPacket(Packet p) {
        // 如果有要注册的watchRegistration
        if (p.watchRegistration != null) {
            // 根据response code进行注册
            p.watchRegistration.register(p.replyHeader.getErr());
        }

        // 如果没有异步回调,  不是异步操作的请求，其会等待packet的finish标识来等待对应的结果
        if (p.cb == null) {
            synchronized (p) {
                p.finished = true;
                p.notifyAll();
            }
        } else {
            //如果有异步回调
            p.finished = true; // 异步操作的则将其加入到waitingEvents队列中等待EventThread的对应线程处理
            eventThread.queuePacket(p);
        }
    }

    private void conLossPacket(Packet p) {
        if (p.replyHeader == null) {
            return;
        }
        switch (state) {
        case AUTH_FAILED:
            p.replyHeader.setErr(KeeperException.Code.AUTHFAILED.intValue());
            break;
        case CLOSED:
            p.replyHeader.setErr(KeeperException.Code.SESSIONEXPIRED.intValue());
            break;
        default:
            p.replyHeader.setErr(KeeperException.Code.CONNECTIONLOSS.intValue());
        }
        finishPacket(p);
    }

    private volatile long lastZxid;

    public long getLastZxid() {
        return lastZxid;
    }

    static class EndOfStreamException extends IOException {
        private static final long serialVersionUID = -5438877188796231422L;

        public EndOfStreamException(String msg) {
            super(msg);
        }
        
        @Override
        public String toString() {
            return "EndOfStreamException: " + getMessage();
        }
    }

    private static class SessionTimeoutException extends IOException {
        private static final long serialVersionUID = 824482094072071178L;

        public SessionTimeoutException(String msg) {
            super(msg);
        }
    }
    
    private static class SessionExpiredException extends IOException {
        private static final long serialVersionUID = -1388816932076193249L;

        public SessionExpiredException(String msg) {
            super(msg);
        }
    }

    private static class RWServerFoundException extends IOException {
        private static final long serialVersionUID = 90431199887158758L;

        public RWServerFoundException(String msg) {
            super(msg);
        }
    }
    
    public static final int packetLen = Integer.getInteger("jute.maxbuffer",
            4096 * 1024);

    /**
     * SendThread是客户端ClientCnxn内部的一个核心I/O调度线程，用于管理客户端与服务端之间的所有网络I/O操作。
     *
     * 在Zookeeper客户端实际运行中，SendThread的作用如下：
     * 1. 维护了客户端与服务端之间的会话生命周期（通过一定周期频率内向服务端发送PING包检测心跳），如果会话周期内客户端与服务端出现TCP连接断开，那么就会自动且透明地完成重连操作。
     * 2. 管理了客户端所有的请求发送和响应接收操作，其将上层客户端API操作转换成相应的请求协议并发送到服务端，并完成对同步调用的返回和异步调用的回调。
     * 3. 将来自服务端的事件传递给EventThread去处理。
     */
    class SendThread extends Thread {
        // 上一次ping的 nano time
        private long lastPingSentNs;
        // 通信层ClientCnxnSocket
        private final ClientCnxnSocket clientCnxnSocket;
        private Random r = new Random(System.nanoTime());
        // 是否第一次connect
        private boolean isFirstConnect = true;

        /**
         * 读取server的回复，进行outgoingQueue以及pendingQueue的相关处理，事件触发等等
         *
         *  1. 处理ping命令,AuthPacket,WatcherEvent,验证sasl并返回
         *  2. 从pendingQueue取出packet进行验证(有顺序保证)
         *  3. 调用finishPacket完成AsyncCallBack处理以及watcher的注册
         *
         * @param incomingBuffer
         * @throws IOException
         */
        void readResponse(ByteBuffer incomingBuffer) throws IOException {
            ByteBufferInputStream bbis = new ByteBufferInputStream( incomingBuffer);
            BinaryInputArchive bbia = BinaryInputArchive.getArchive(bbis);
            //反序列化出 回复头
            ReplyHeader replyHdr = new ReplyHeader();

            replyHdr.deserialize(bbia, "header");
            if (replyHdr.getXid() == -2) {
                return;
            }
            if (replyHdr.getXid() == -4) {
                if(replyHdr.getErr() == KeeperException.Code.AUTHFAILED.intValue()) {
                    state = States.AUTH_FAILED;
                    // 加入eventThread
                    eventThread.queueEvent( new WatchedEvent(Watcher.Event.EventType.None,  Watcher.Event.KeeperState.AuthFailed, null) );
                }
                return;
            }
            // -1代表通知类型 即WatcherEvent
            if (replyHdr.getXid() == -1) {
                // 这里会接收事件通知，反序列化
                WatcherEvent event = new WatcherEvent();
                // 反序列化WatcherEvent
                event.deserialize(bbia, "response");

                // 把serverPath转化成clientPath
                if (chrootPath != null) {
                    String serverPath = event.getPath();
                    if(serverPath.compareTo(chrootPath)==0) {
                        event.setPath("/");
                    } else if (serverPath.length() > chrootPath.length()) {
                        event.setPath(serverPath.substring(chrootPath.length()));
                    } else {
                    	LOG.warn("Got server path " + event.getPath() + " which is too short for chroot path " + chrootPath);
                    }
                }

                // WatcherEvent还原成WatchedEvent
                WatchedEvent we = new WatchedEvent(event);

                // 核心方法：加入eventThread
                eventThread.queueEvent( we );
                return;
            }

            if (clientTunneledAuthenticationInProgress()) {
                GetSASLRequest request = new GetSASLRequest();
                request.deserialize(bbia,"token");
                zooKeeperSaslClient.respondToServer(request.getToken(), ClientCnxn.this);
                return;
            }

            // auth和ping以及正在处理的sasl没有加入pendingQueue,
            // 触发的watch也没有在pendingQueue中(是server主动发过来的),
            // 而AsyncCallBack在pendingQueue中(见finishPacket)
            Packet packet;
            synchronized (pendingQueue) {
                if (pendingQueue.size() == 0) {
                    throw new IOException("Nothing in the queue, but got " + replyHdr.getXid());
                }
                // 得到了response，从pendingQueue中移除
                packet = pendingQueue.remove();
            }
            try {
                if (packet.requestHeader.getXid() != replyHdr.getXid()) {
                    packet.replyHeader.setErr(KeeperException.Code.CONNECTIONLOSS.intValue());
                    throw new IOException("Xid out of order. Got Xid "
                            + replyHdr.getXid() + " with err " +
                            + replyHdr.getErr() + " expected Xid "
                            + packet.requestHeader.getXid() + " for a packet with details: "
                            + packet );
                }
                // 设置replyHeader
                packet.replyHeader.setXid(replyHdr.getXid());
                packet.replyHeader.setErr(replyHdr.getErr());
                packet.replyHeader.setZxid(replyHdr.getZxid());
                if (replyHdr.getZxid() > 0) {
                    lastZxid = replyHdr.getZxid();
                }
                if (packet.response != null && replyHdr.getErr() == 0) {
                    packet.response.deserialize(bbia, "response");
                }
            } finally {
                finishPacket(packet);
            }
        }

        // 将当前客户端连接状态置为States.CONNECTING，并初始化了ClientCnxnSocket
        SendThread(ClientCnxnSocket clientCnxnSocket) {
            super(makeThreadName("-SendThread()"));
            state = States.CONNECTING;
            this.clientCnxnSocket = clientCnxnSocket;
            setUncaughtExceptionHandler(uncaughtExceptionHandler);
            setDaemon(true);
        }

        /**
         * Used by ClientCnxnSocket
         * @return
         */
        ZooKeeper.States getZkState() {
            return state;
        }

        ClientCnxnSocket getClientCnxnSocket() {
            return clientCnxnSocket;
        }

        /**
         * 主要连接函数，用于将watches和authData传给server，允许clientCnxnSocket可读写
         *
         * 如果瞬间连上，就直接调用。 否则的话就等到sc.finishConnect()再调用。
         * 这个函数完成了一些watches和authData的传递以及允许更改SelectionKey，允许clientCnxnSocket可读写，
         *
         *   1. 根据之前是否连接过设置sessId以及生成ConnectRequest
         *   2. 根据disableAutoWatchReset将已有的watches和authData以及放入outgoingQueue准备发送
         *   3. 允许clientCnxnSocket可读写,表示和server准备IO
         *
         * 主要注意:
         *   1. sessId:如果之前连接过，那么重连用之前的sessionId，否则默认0,重连参见ClientCnxn.SendThread#startConnect的调用
         *   2. 什么时候连接会有watches需要去注册? 重连且disableAutoWatchReset为false的时候
         *   3. ConnectRequest是放在outgoingQueue第一个的，确保最先发出去的是连接请求(保证了第一个response是被ClientCnxnSocket#readConnectResult处理)
         *
         * @throws IOException
         */
        void primeConnection() throws IOException {
            LOG.info("Socket connection established to " + clientCnxnSocket.getRemoteSocketAddress() + ", initiating session");
            isFirstConnect = false;
            // 如果之前见过读写server就设置sessionId,否则默认0
            long sessId = (seenRwServerBefore) ? sessionId : 0;
            // 生成ConnectRequest
            ConnectRequest conReq = new ConnectRequest(0, lastZxid, sessionTimeout, sessId, sessionPasswd);

            synchronized (outgoingQueue) {
                if (!disableAutoWatchReset) {
                    List<String> dataWatches = zooKeeper.getDataWatches();    // 监听数据变化
                    List<String> existWatches = zooKeeper.getExistWatches();  // 监听节点是否存在
                    List<String> childWatches = zooKeeper.getChildWatches();  // 监听子节点变化
                    if (!dataWatches.isEmpty() || !existWatches.isEmpty() || !childWatches.isEmpty()) {
                        SetWatches sw = new SetWatches(lastZxid,
                                                prependChroot(dataWatches),
                                                prependChroot(existWatches),
                                                prependChroot(childWatches)); // 设置watches
                        RequestHeader h = new RequestHeader();
                        h.setType(ZooDefs.OpCode.setWatches);
                        h.setXid(-8);
                        Packet packet = new Packet(h, new ReplyHeader(), sw, null, null);
                        outgoingQueue.addFirst(packet); // 加入发送队列， 封装了packet, 里面封装了一堆你要施加的监听器
                    }
                }

                for (AuthData id : authInfo) {
                    // authInfo加入发送队列
                    outgoingQueue.addFirst(new Packet(new RequestHeader(-4, OpCode.auth), null, new AuthPacket(0, id.scheme, id.data), null, null));
                }
                // 封装一个ConnectRequest, 放入 packet queue里去了
                // ConnectRequest确保在发送队列的第一个
                outgoingQueue.addFirst(new Packet(null, null, conReq, null, null, readOnly));
            }
            // 开启读写，这样outgoingQueue内容就可以发出去了
            clientCnxnSocket.enableReadWriteOnly();
        }

        /**
         * 根据clientPath以及chrootPath得到serverPath
         * @param paths
         * @return
         */
        private List<String> prependChroot(List<String> paths) {
            if (chrootPath != null && !paths.isEmpty()) {
                for (int i = 0; i < paths.size(); ++i) {
                    String clientPath = paths.get(i);
                    String serverPath;
                    if (clientPath.length() == 1) {
                        serverPath = chrootPath;
                    } else {
                        serverPath = chrootPath + clientPath;
                    }
                    paths.set(i, serverPath);
                }
            }
            return paths;
        }

        /**
         * ping命令，记录发出时间，生成请求，加入outgoingQueue待发送
         *
         * 注意一点，在run方法会将outgoingQueue的内容发送出去，
         * 在ClientCnxnSocketNIO#doIO中， ping命令的packet是没有进入pendingQueue的
         */
        private void sendPing() {
            lastPingSentNs = System.nanoTime();
            RequestHeader h = new RequestHeader(-2, OpCode.ping);
            queuePacket(h, null, null, null, null, null, null, null, null);
        }

        /** 读写server地址 */
        private InetSocketAddress rwServerAddress = null;
        /** 最短ping 读写server的 timeout时间 */
        private final static int minPingRwTimeout = 100;
        /** 最长ping 读写server的 timeout时间 */
        private final static int maxPingRwTimeout = 60000;
        /** 默认ping 读写server的 timeout时间 */
        private int pingRwTimeout = minPingRwTimeout;
        /** sasl登录失败 */
        private boolean saslLoginFailed = false;

        /**
         * 开始连接，主要是和server的socket完成connect和accept
         *
         *     1. 根据hostProvider或者已经设置的读写服务器地址确定server 地址
         *     2. sasl相关处理
         *     3. 调用clientCnxnSocket.connect
         * @throws IOException
         */
        private void startConnect() throws IOException {
            state = States.CONNECTING;

            InetSocketAddress addr;
            if (rwServerAddress != null) {
                // 有rwServerAddress 就设置
                addr = rwServerAddress;
                rwServerAddress = null;
            } else {
                // 没有就从服务器地址列表取出来一个
                addr = hostProvider.next(1000);
            }

            // 设置线程名字
            setName(getName().replaceAll("\\(.*\\)", "(" + addr.getHostName() + ":" + addr.getPort() + ")"));
            try {
                zooKeeperSaslClient = new ZooKeeperSaslClient("zookeeper/"+addr.getHostName());
            } catch (LoginException e) {
                eventThread.queueEvent(new WatchedEvent( Watcher.Event.EventType.None, Watcher.Event.KeeperState.AuthFailed, null) );
                saslLoginFailed = true;
            }

            logStartConnect(addr);
            // 核心方法：socket连接
            clientCnxnSocket.connect(addr);
        }

        private void logStartConnect(InetSocketAddress addr) {
            String msg = "Opening socket connection to server " + addr;
            if (zooKeeperSaslClient != null) {
              msg += ". " + zooKeeperSaslClient.getConfigStatus();
            }
            LOG.info(msg);
        }

        private static final String RETRY_CONN_MSG = ", closing socket connection and attempting reconnect";

        /**
         * 线程方法，完成了连接验证，超时检测，ping命令，以及网络IO
         * 1. clientCnxnSocket相关初始化,
         * 2. 不断检测clientCnxnSocket是否和服务器处于连接状态, 没有连接则进行连接,
         * 3. 检测是否超时：当处于连接状态时，检测是否读超时，当处于未连接状态时，检测是否连接超时,
         * 4. 不断的发送ping通知，服务器端每接收到ping请求，就会从当前时间重新计算session过期时间，所以当客户端按照一定时间间隔不断的发送ping请求，就能保证客户端的session不会过期,
         * 5. 如果当前是只读的话，不断去找有没有支持读写的server,
         * 6. 不断进行IO操作，发送请求队列中的请求和读取服务器端的响应数据,
         *    - clientCnxnSocket.doTransport(to, pendingQueue, outgoingQueue, ClientCnxn.this); //在特定时间内，根据两个queue进行网络传输
         * 7. !state.isAlive()时，进行相关清理工作,
         */
        @Override
        public void run() {
            // 1. clientCnxnSocket相关初始化
            // 初始化clientCnxnSocket.sendThread和clientCnxnSocket.sessionId，后者此时为0
            clientCnxnSocket.introduce(this,sessionId); //引入原来的sessionId,用于做重连操作
            // 更新clientCnxnSocket.now为当前时间
            clientCnxnSocket.updateNow();
            // 更新上一次发送和收到的时间为当前时间
            clientCnxnSocket.updateLastSendAndHeard();
            int to;
            long lastPingRwServer = System.currentTimeMillis();

            // 2. 不断检测clientCnxnSocket是否和服务器处于连接状态,没有连接则进行连接
            while (state.isAlive()) { // 不是Closed和Auth_faild状态
                try {
                    // 如果clientCnxnSocket的SelectionKey为null,说明底层连接已和服务端断开
                    if (!clientCnxnSocket.isConnected()) {
                        // 如果不是第一次连接且连接断开， 很可能是zk服务器宕机了，重新进行连接
                        if(!isFirstConnect){
                            try {
                                Thread.sleep(r.nextInt(1000));
                            } catch (InterruptedException e) {
                                LOG.warn("Unexpected exception", e);
                            }
                        }
                        if (closing || !state.isAlive()) {
                            break;
                        }
                        //核心方法：非客户端主动关闭或认证失败，连接断开时客户端会重新连接服务端
                        //也是第一次连接的入口
                        startConnect(); // 开始连接,并且切换状态为CONNECTIONG,主要是调用clientCnxnSocket对应的connect方法
                        clientCnxnSocket.updateLastSendAndHeard(); //更新对应的发送和接收时间
                    }

                    // 3. 检测是否超时：当处于连接状态时，检测是否读超时，当处于未连接状态时，检测是否连接超时
                    // 检测是否超时,分为读超时和连接超时
                    if (state.isConnected()) {
                        if (zooKeeperSaslClient != null) {
                            boolean sendAuthEvent = false;
                            if (zooKeeperSaslClient.getSaslState() == ZooKeeperSaslClient.SaslState.INITIAL) {
                                try {
                                    zooKeeperSaslClient.initialize(ClientCnxn.this);
                                } catch (SaslException e) {
                                   LOG.error("SASL authentication with Zookeeper Quorum member failed: " + e);
                                    state = States.AUTH_FAILED;
                                    sendAuthEvent = true;
                                }
                            }
                            // 这个get方法会切换zooKeeperSaslClient的状态，则SaslAuthenticated只会出现一次
                            KeeperState authState = zooKeeperSaslClient.getKeeperState();
                            if (authState != null) {
                                if (authState == KeeperState.AuthFailed) {
                                    state = States.AUTH_FAILED;
                                    sendAuthEvent = true;
                                } else {
                                    if (authState == KeeperState.SaslAuthenticated) {
                                        sendAuthEvent = true;
                                    }
                                }
                            }

                            if (sendAuthEvent == true) {
                                eventThread.queueEvent(new WatchedEvent( Watcher.Event.EventType.None,  authState,null));
                            }
                        }
                        // 如果已经连接上，预计读超时时间 - 距离上次读已经过去的时间
                        to = readTimeout - clientCnxnSocket.getIdleRecv();
                    } else {
                        // 如果没连接上, 重置连接可用时间, 预计连接时间 - 上次读已经过去的时间
                        to = connectTimeout - clientCnxnSocket.getIdleRecv();
                    }

                    // 代表读超时或连接超时
                    if (to <= 0) {
                        throw new SessionTimeoutException( "Client session timed out, have not heard from server in " + clientCnxnSocket.getIdleRecv() + "ms" + " for sessionid 0x"  + Long.toHexString(sessionId));
                    }

                    // 4. 不断的发送ping通知, 服务器端每接收到ping请求，就会从当前时间重新计算session过期时间，所以当客户端按照一定时间间隔不断的发送ping请求，就能保证客户端的session不会过期
                    if (state.isConnected()) {
                        int timeToNextPing = readTimeout / 2 - clientCnxnSocket.getIdleSend();
                        // readTimeout已经过了近一半的时间，或者距离上次发送请求已过过了10s
                        // 已经有多长时间没有发送消息到服务端去了,比如5秒
                        // 20s - 5s = 15s
                        if (timeToNextPing <= 0) { //最少MAX_SEND_PING_INTERVAL发送一次ping
                            // 客户端一定每个一段时间发送心跳
                            sendPing();
                            clientCnxnSocket.updateLastSend();
                        } else {
                            // 如果预计下次ping的时间 < 实际距离下次ping的时间
                            if (timeToNextPing < to) {
                                to = timeToNextPing;
                            }
                        }
                    }

                    // 5. 如果当前是只读的话，不断去找有没有支持读写的server
                    if (state == States.CONNECTEDREADONLY) {  //ping找不是只读的对应的server
                        long now = System.currentTimeMillis();
                        int idlePingRwServer = (int) (now - lastPingRwServer);
                        if (idlePingRwServer >= pingRwTimeout) {
                            lastPingRwServer = now;
                            idlePingRwServer = 0;
                            pingRwTimeout = Math.min(2*pingRwTimeout, maxPingRwTimeout);
                            pingRwServer();
                        }
                        to = Math.min(to, pingRwTimeout - idlePingRwServer);
                    }

                    // 最核心的代码，
                    // 底层封装的ClientCnxnSocket把outgoingQueue里的数据发送出去
                    // 6. 不断进行IO操作，发送请求队列中的请求和读取服务器端的响应数据
                    clientCnxnSocket.doTransport(to, pendingQueue, outgoingQueue, ClientCnxn.this);
                } catch (Throwable e) {
                    // 一旦你的zk服务器断开，所有的zk客户端，都会关闭连接，清空自己的监听器，然后重新去连接一台zk服务器。
                    // 每次触发事件之后，你还得重新去进行一下注册，要不然监听器是一次性的，下次就没有了。
                    // 如果跟zk服务器的连接断开了，此时监听器就都没有了，你还是得在创建ZooKeeper的时候给一个默认的监听器。
                    // 如果你收到了连接断开的通知之后，此时你应该知道zk服务器已经断开了，此时你应该等待，他会自动重新去连接其他的zk服务器，如果连接成功了，还会给你一个通知，告诉你是连接已经重新建立了，重新去进行监听器的注册。

                    // 如果zk宕机了，此时doIO(){socket.write()} 一定会抛出IOException,
                    // 此时就可以认为是zk服务器可能是宕机，也可能是网络出现了问题，反正就是跟这台zk服务器失联了。
                    if (closing) {
                        break;
                    } else {
                        if (e instanceof SessionExpiredException) {
                            LOG.info(e.getMessage() + ", closing socket connection");
                        } else if (e instanceof SessionTimeoutException) {
                            LOG.info(e.getMessage() + RETRY_CONN_MSG);
                        } else if (e instanceof EndOfStreamException) {
                            LOG.info(e.getMessage() + RETRY_CONN_MSG);
                        } else if (e instanceof RWServerFoundException) {
                            LOG.info(e.getMessage());
                        } else {
                            LOG.warn(  "Session 0x" + Long.toHexString(getSessionId()) + " for server " + clientCnxnSocket.getRemoteSocketAddress() + ", unexpected error" + RETRY_CONN_MSG, e);
                        }
                        // 说明线程出现异常，清空pendingQueue和outgoingQueue，需要等待客户端重新连接
                        cleanup();
                        if (state.isAlive()) {
                            // 核心方法：在这里会发布一个事件，Disconnected
                            // 最早在创建zookeeper的时候加入了一个默认的监听器
                            eventThread.queueEvent(new WatchedEvent( Event.EventType.None, Event.KeeperState.Disconnected, null));
                        }
                        // 重新更新它的底层网络连接初始化的一个时间
                        clientCnxnSocket.updateNow();
                        // 重新初始化底层网络连接里的最近一次发送请求和接收请求的时间
                        clientCnxnSocket.updateLastSendAndHeard();
                    }
                }
            }
            // 7. !state.isAlive()时，进行相关清理工作
            cleanup();
            clientCnxnSocket.close(); // 关闭socket
            if (state.isAlive()) {
                eventThread.queueEvent(new WatchedEvent(Event.EventType.None, Event.KeeperState.Disconnected, null));
            }
            ZooTrace.logTraceMessage(LOG, ZooTrace.getTextTraceLevel(), "SendThread exitedloop.");
        }

        /**
         * ping读写server.  找到读写server，更新rwServerAddress
         * @throws RWServerFoundException
         */
        private void pingRwServer() throws RWServerFoundException {
            String result = null;
            InetSocketAddress addr = hostProvider.next(0);
            LOG.info("Checking server " + addr + " for being r/w." +  " Timeout " + pingRwTimeout);

            try {
                Socket sock = new Socket(addr.getHostName(), addr.getPort());
                sock.setSoLinger(false, -1);
                sock.setSoTimeout(1000);
                sock.setTcpNoDelay(true);
                sock.getOutputStream().write("isro".getBytes());
                sock.getOutputStream().flush();
                sock.shutdownOutput();
                BufferedReader br = new BufferedReader( new InputStreamReader(sock.getInputStream()));
                result = br.readLine();
                sock.close();
                br.close();
            } catch (ConnectException e) {
            } catch (IOException e) {
                LOG.warn("Exception while seeking for r/w server " + e.getMessage(), e);
            }

            // 如果更新了rwServerAddress 会抛异常，run方法处理RWServerFoundException异常，会清理后cleanup()进行重连,来让client重连到读写server上去
            if ("rw".equals(result)) {
                pingRwTimeout = minPingRwTimeout;
                rwServerAddress = addr;
                throw new RWServerFoundException("Majority server found at "  + addr.getHostName() + ":" + addr.getPort());
            }
        }

        /**
         * socket清理以及通知两个queue失去连接 以及 清理两个队列
         *
         * 这是客户端主动断开连接的过程是非常经典的，非常值得我们学习，对我们编写生产级别的代码非常有帮助，
         * 1. 对代码进行良好的封装，定义生产级的协议、状态码、输入输出的参数的标识，
         * 2. 对大量的异常进行良好的捕获和处理，
         */
        private void cleanup() {
            // 1. 客户端主动断开跟服务器的连接
            clientCnxnSocket.cleanup();
            // 2. 对所有已经发送出去，正在等待响应的请求，都进行完成，标识其失败
            synchronized (pendingQueue) {
                for (Packet p : pendingQueue) {
                    conLossPacket(p);
                }
                pendingQueue.clear();
            }
            // 3. 对所有还没有发送出去的请求，都进行完成，标识其失败
            synchronized (outgoingQueue) {
                for (Packet p : outgoingQueue) {
                    conLossPacket(p);
                }
                outgoingQueue.clear();
            }
        }

        /**
         * 读取server的connect response后，设置相关参数
         * 
         * @param _negotiatedSessionTimeout
         * @param _sessionId
         * @param _sessionPasswd
         * @param isRO
         * @throws IOException
         */
        void onConnected(int _negotiatedSessionTimeout, long _sessionId, byte[] _sessionPasswd, boolean isRO) throws IOException {
            negotiatedSessionTimeout = _negotiatedSessionTimeout;
            if (negotiatedSessionTimeout <= 0) {
                state = States.CLOSED;

                eventThread.queueEvent(new WatchedEvent( Watcher.Event.EventType.None, Watcher.Event.KeeperState.Expired, null));
                eventThread.queueEventOfDeath();
                throw new SessionExpiredException( "Unable to reconnect to ZooKeeper service, session 0x" + Long.toHexString(sessionId) + " has expired");
            }
            // 如果client不允许只读，但是目前是只读
            if (!readOnly && isRO) {
                LOG.error("Read/write client got connected to read-only server");
            }
            readTimeout = negotiatedSessionTimeout * 2 / 3;
            connectTimeout = negotiatedSessionTimeout / hostProvider.size();
            // 更新hostProvider循环列表的index
            hostProvider.onConnected();
            sessionId = _sessionId;
            sessionPasswd = _sessionPasswd;
            // 根据isRO设置state
            state = (isRO) ? States.CONNECTEDREADONLY : States.CONNECTED;
            // 是否见过读写server
            seenRwServerBefore |= !isRO;
            KeeperState eventState = (isRO) ? KeeperState.ConnectedReadOnly : KeeperState.SyncConnected;
            // 加入watcherEvent
            eventThread.queueEvent(new WatchedEvent( Watcher.Event.EventType.None, eventState, null));
        }

        void close() {
            state = States.CLOSED;
            clientCnxnSocket.wakeupCnxn();
        }

        void testableCloseSocket() throws IOException {
            clientCnxnSocket.testableCloseSocket();
        }

        /**
         * 是否在验证sasl
         * @return
         */
        public boolean clientTunneledAuthenticationInProgress() {
            // 1. SASL 登录失败.
            if (saslLoginFailed == true) {
                return false;
            }

            // 2. SendThread 还没有创建 authenticating 对象, 因此 authentication 在 (在生命周期的最初阶段) 进行中.
            if (zooKeeperSaslClient == null) {
                return true;
            }

            // 3. authenticating 对象存在 , 所以询问它的进展.
            return zooKeeperSaslClient.clientTunneledAuthenticationInProgress();
        }

        public void sendPacket(Packet p) throws IOException {
            clientCnxnSocket.sendPacket(p);
        }
    }

    /**
     * Shutdown the send/event threads. This method should not be called
     * directly - rather it should be called as part of close operation. This
     * method is primarily here to allow the tests to verify disconnection
     * behavior.
     */
    public void disconnect() {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Disconnecting client for session: 0x"
                      + Long.toHexString(getSessionId()));
        }

        sendThread.close();
        eventThread.queueEventOfDeath();
    }

    /**
     * Close the connection, which includes; send session disconnect to the
     * server, shutdown the send/event threads.
     *
     * @throws IOException
     */
    public void close() throws IOException {
        if (LOG.isDebugEnabled()) {
            LOG.debug("Closing client for session: 0x"
                      + Long.toHexString(getSessionId()));
        }

        try {
            RequestHeader h = new RequestHeader();
            h.setType(ZooDefs.OpCode.closeSession);

            submitRequest(h, null, null, null);
        } catch (InterruptedException e) {
            // ignore, close the send/event threads
        } finally {
            disconnect();
        }
    }

    private int xid = 1;

    private volatile States state = States.NOT_CONNECTED;

    /*
     * getXid() is called externally by ClientCnxnNIO::doIO() when packets are sent from the outgoingQueue to
     * the server. Thus, getXid() must be public.
     */
    synchronized public int getXid() {
        return xid++;
    }

    public ReplyHeader submitRequest(RequestHeader h, Record request, Record response, WatchRegistration watchRegistration) throws InterruptedException {
        // 生成回复头
        ReplyHeader r = new ReplyHeader();

        Packet packet = queuePacket(h, r, request, response, null, null, null, null, watchRegistration);

        synchronized (packet) {
            // 如果packet没有处理完,就一直等着
            // 直到你的请求发送完毕，同时接收到响应之后，这里才会退出
            while (!packet.finished) {
                packet.wait();
            }
        }
        return r;
    }

    public void enableWrite() {
        sendThread.getClientCnxnSocket().enableWrite();
    }

    public void sendPacket(Record request, Record response, AsyncCallback cb, int opCode)
    throws IOException {
        // Generate Xid now because it will be sent immediately,
        // by call to sendThread.sendPacket() below.
        int xid = getXid();
        RequestHeader h = new RequestHeader();
        h.setXid(xid);
        h.setType(opCode);

        ReplyHeader r = new ReplyHeader();
        r.setXid(xid);

        Packet p = new Packet(h, r, request, response, null, false);
        p.cb = cb;
        sendThread.sendPacket(p);
    }

    Packet queuePacket(RequestHeader h, ReplyHeader r, Record request, Record response, AsyncCallback cb, String clientPath, String serverPath, Object ctx, WatchRegistration watchRegistration) {
        Packet packet = null;

        synchronized (outgoingQueue) {
            packet = new Packet(h, r, request, response, watchRegistration);
            packet.cb = cb;
            packet.ctx = ctx;
            packet.clientPath = clientPath;
            packet.serverPath = serverPath;
            if (!state.isAlive() || closing) {
                conLossPacket(packet);
            } else {
                if (h.getType() == OpCode.closeSession) {
                    closing = true;
                }
                // 将packet加入到 outgoingQueue队列中
                outgoingQueue.add(packet);
            }
        }
        // 唤醒底层网络通信的组件
        sendThread.getClientCnxnSocket().wakeupCnxn();
        return packet;
    }

    public void addAuthInfo(String scheme, byte auth[]) {
        if (!state.isAlive()) {
            return;
        }
        authInfo.add(new AuthData(scheme, auth));
        queuePacket(new RequestHeader(-4, OpCode.auth), null,
                new AuthPacket(0, scheme, auth), null, null, null, null, null, null);
    }

    States getState() {
        return state;
    }
}
